package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo11Join {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Join算子演示")
    val context = new SparkContext(conf)
    //====================================================

    //两个kv类型的RDD之间的关联
    //通过scala中的集合构建RDD
    val rdd1: RDD[(String, String)] = context.parallelize(
      List(
        ("1001", "尚平"),
        ("1002", "丁义杰"),
        ("1003", "徐昊宇"),
        ("1004", "包旭"),
        ("1005", "朱大牛"),
        ("1006","汪权")
      )
    )

    val rdd2: RDD[(String, String)] = context.parallelize(
      List(
        ("1001", "崩坏"),
        ("1002", "原神"),
        ("1003", "王者"),
        ("1004", "修仙"),
        ("1005", "学习"),
        ("1007", "敲代码")
      )
    )

    /**
     *  内连接：join
     *  左连接：leftJoin
     *  右连接：rightJoin
     *  全连接：fullJoin
     */

    //内连接
//    val innerJoinRDD: RDD[(String, (String, String))] = rdd1.join(rdd2)
//    //加工一下RDD
//    val innerJoinRDD2: RDD[(String, String, String)] = innerJoinRDD.map {
//      case (id: String, (name: String, like: String)) => (id, name, like)
//    }
//    innerJoinRDD2.foreach(println)

    //左连接
    val leftJoinRDD: RDD[(String, (String, Option[String]))] = rdd1.leftOuterJoin(rdd2)
    //加工一下RDD
    val leftJoinRDD2: RDD[(String, String, String)] = leftJoinRDD.map {
      case (id: String, (name: String, Some(like))) => (id, name, like)
      case (id: String, (name: String, None)) => (id, name, "无爱好")
    }
    leftJoinRDD2.foreach(println)

    println("=================================")
    //右连接自己试
    //TODO:自己试右连接

    //全连接
    val fullJoinRDD: RDD[(String, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)
    //加工一下RDD
    val fullJoinRDD2: RDD[(String, String, String)] = fullJoinRDD.map {
      case (id: String, (Some(name), Some(like))) => (id, name, like)
      case (id: String, (Some(name), None)) => (id, name, "无爱好")
      case (id: String, (None, Some(like))) => (id, "无姓名", like)
    }
    fullJoinRDD2.foreach(println)




  }

}
